package com.shujia.spark.core

import org.apache.spark.{SparkConf, SparkContext}

import scala.io.Source

object Demo21Bro {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("app")

    val sc = new SparkContext(conf)

    val rdd = sc.textFile("spark/data/score.txt")

    println(rdd.getNumPartitions)

    //重分区，会产生shuffle,  提高任务的并行度
    val scoreRDD = rdd.repartition(4)
    println(scoreRDD.getNumPartitions)

    val studentList = Source.fromFile("spark/data/students.txt").getLines().toList

    val studentMap = studentList.map(line => (line.split(",")(0), line)).toMap

    /*    val joinRDD = scoreRDD.map(line => {
          val split = line.split(",")
          val id = split(0)
          //通过学号去studentMap中去出学生信息
          //studentMap ： 每一个分区都会产生一个变量副本
          val studentInfo = studentMap.getOrElse(id, "默认值")
          studentInfo + "\t" + line
        })

        joinRDD.foreach(println)*/


    /**
      * 广播变量
      *
      * 只能在Driver端进行定义，只能在Executor端进行使用，
      * 不能在Executor端改变变量的数据，就算改变了，在其他Executor中也不会生效
      * 广播的变量必须可以序列化，因为这个变量要在网络中传输
      * rdd 不能广播
      *
      *
      * 通过广播变量可以实现mapjoin,  将小表加载到内存， 大表关联小表不会产生shuffle
      * 小表的数据量不能超过Driver和Executor的内存，  而且不能太大  不能大于10G
      */

    //1、在Driver端对一个scala的变量进行广播
    val studentMapBro = sc.broadcast(studentMap)

    scoreRDD.map(line => {
      val split = line.split(",")
      val id = split(0)

      //2、在Executor中使用关播变量
      val bro = studentMapBro.value
      val studentInfo = bro.getOrElse(id, "默认值")

      studentInfo + "\t" + line


    }).foreach(println)


  }

}
